package cn.gwm.flink.streaming.strategy.vehiclepredict;

import cn.gwm.flink.streaming.constant.ChargePredictFields;
import org.apache.flink.core.io.SimpleVersionedSerializer;
import org.apache.flink.streaming.api.functions.sink.filesystem.BucketAssigner;
import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.SimpleVersionedStringSerializer;
import org.apache.flink.util.Preconditions;

import java.time.ZoneId;

/**
 * @author : GW00280745
 * @date : 2022/7/11 10:22
 * @description : 自定义分区。例如使用事件时间：event time = partition name
 * 因同事规定pom文件不能动所以目前只能用row行存储。基于此尽量不要提交和源码不一致的东西，引起误解。
 * 等规范设定好后再按规范修改。
 */
public class ChargeBucketAssigner<IN>  implements BucketAssigner<IN, String> {
    private static final long serialVersionUID = 1L;

    private static final String DEFAULT_FORMAT_STRING = "yyyy-MM-dd";

    private final String formatString;

    private final ZoneId zoneId;

    public ChargeBucketAssigner() {
        this(DEFAULT_FORMAT_STRING, ZoneId.systemDefault());
    }

    public ChargeBucketAssigner(String formatString, ZoneId zoneId) {
        this.formatString = Preconditions.checkNotNull(formatString);
        this.zoneId = Preconditions.checkNotNull(zoneId);
    }
    private String partStr = "dt=";
    @Override
    public String getBucketId(IN element, Context context) {
        if (element instanceof String){
            String line = element.toString();
            String[] sp = line.split(ChargePredictFields.ins().charTab);
            String s = sp[sp.length - 1];
            return partStr+s;
        }
        else {
            return "error";
        }


    }

    @Override
    public SimpleVersionedSerializer<String> getSerializer() {
        return SimpleVersionedStringSerializer.INSTANCE;
    }

    @Override
    public String toString() {
        return "ChargeBucketAssigner{"
                + "formatString='"
                + formatString
                + '\''
                + ", zoneId="
                + zoneId
                + '}';
    }
}
